//one file per object
package plain

import (
	"encoding/binary"
	"errors"
	"fmt"
	"os"
	"reflect"
	.	"store"
	"syscall"
)

type Plain struct {
	Dir          string
	Id           uint32
	repEntryChan chan ReplicaEntry
	keyListErr   error
	done         chan bool
}

const (
	PlainFilenameFormat = "%s/%d.plain"
)

func NewPlain(dir string, id uint32) *Plain {
	pln := new(Plain)
	pln.Dir = dir
	pln.Id = id
	pln.repEntryChan = make(chan ReplicaEntry, 10)
	pln.done = make(chan bool)
	go pln.writeKeyListFiles()
	return pln
}

func (p *Plain) Put(key uint64, otherStoreIds []uint32, realname string, length uint64) error {
	file, err := createPlainFile(p.Dir, key)
	if err != nil {
		return err
	}
	defer file.Close()
	buf := make([]byte, len(realname)+4)
	binary.LittleEndian.PutUint32(buf, uint32(len(realname)))
	_, err = file.Write(buf)

	for _, storeId := range otherStoreIds {
		var re ReplicaEntry
		re.Id = storeId
		re.Key = key
		p.repEntryChan <- re
	}
	return err
}

func (p *Plain) Write(key uint64, data []byte, offset uint64) error {
	file, err := openPlainFile(p.Dir, key)
	if err != nil {
		return err
	}
	defer file.Close()
	_, err = file.WriteAt(data, int64(offset))
	return err
}

func (p *Plain) Get(key uint64, offset uint64, size uint32) (data []byte, err error) {
	file, err := openPlainFile(p.Dir, key)
	if err != nil {
		return nil, err
	}
	defer file.Close()
	data = make([]byte, size)
	_, err = file.ReadAt(data, int64(offset))
	return
}

func (p *Plain) GetName(key uint64) (string, error) {
	file, err := openPlainFile(p.Dir, key)
	if err != nil {
		return "", err
	}
	defer file.Close()
	sizeBuf := make([]byte, 4)
	_, err = file.Read(sizeBuf)
	if err != nil {
		return "", err
	}
	size := binary.LittleEndian.Uint32(sizeBuf)
	nameBuf := make([]byte, size)
	_, err = file.Read(nameBuf)
	if err != nil {
		return "", err
	}
	return string(nameBuf), nil
}

func (p *Plain) Delete(key uint64) error {
	fn := fmt.Sprintf(PlainFilenameFormat, p.Dir, key)
	return os.Remove(fn)
}

func (p *Plain) SpaceInfo() (total, free uint64) {
	statfs := new(syscall.Statfs_t)
	syscall.Statfs(p.Dir, statfs)
	statfsValue := reflect.ValueOf(statfs).Elem()
	bsize := uint64(statfsValue.FieldByName("Bsize").Interface().(int64))
	blocks := statfsValue.FieldByName("Blocks").Interface().(uint64)
	availBlocks := statfsValue.FieldByName("Bavail").Interface().(uint64)
	total = bsize * blocks
	free = bsize * availBlocks
	return
}

func openPlainFile(dir string, key uint64) (file *os.File, err error) {
	fn := fmt.Sprintf(PlainFilenameFormat, dir, key)
	return os.Open(fn)
}
func openReplicaFile(filename, meta string) (*os.File, error) {
	file, err := os.OpenFile(filename, os.O_RDWR, 0666)
	if err != nil {
		return nil, err
	}
	buf := make([]byte, AlignSize)
	_, err = file.Read(buf)
	if err != nil {
		return nil, err
	}
	if len(meta) > AlignSize || string(buf[:len(meta)]) != meta {
		return nil, errors.New("plain:can not open replica file, meta is incorrect.")
	}
	return file, nil
}

//create a file that is not existed.
func createPlainFile(dir string, key uint64) (file *os.File, err error) {
	fn := fmt.Sprintf(PlainFilenameFormat, dir, key)
	file, err = os.Open(fn)
	if file != nil { //
		file.Close()
		return nil, ErrDuplicatedKey
	}
	return os.Create(fn)
}

func createKeyListFile(filename, meta string) (*os.File, error) {
	file, err := os.Create(filename)
	if err != nil {
		return nil, err
	}
	buf := make([]byte, AlignSize)
	copy(buf, []byte(meta))
	_, err = file.Write(buf)
	if err != nil {
		return nil, err
	}
	return file, nil
}

// run as goroutine
func (pln *Plain) writeKeyListFiles() {
	buf := make([]byte, 8)
	keyListFiles := make(map[uint32]*os.File)
	for ri := range pln.repEntryChan {
		file, ok := keyListFiles[ri.id]
		var err error
		if !ok {
			file, err = pln.openKeyListFile(ri.id)
			if file != nil {
				keyListFiles[ri.id] = file
			}
		}
		if err != nil {
			pln.keyListErr = err
		}
		binary.LittleEndian.PutUint64(buf, ri.key)
		_, err = file.Write(buf)
		if err != nil {
			pln.keyListErr = err
		}
	}
	for _, file := range keyListFiles {
		file.Close()
	}
	pln.done <- true
}

func (p *Plain) openReplicaEntryFile(storeId uint32) (file *os.File, err error) {
	filename := fmt.Sprintf("%s/%d.plni", p.Dir, storeId)
	meta := fmt.Sprintf("JFS/v1/plain/%d/replica/%d\n", p.Id, storeId)
	file, err = OpenFile(filename, meta)
	if os.IsNotExist(err) {
		file, err = CreateFile(filename, meta)
	}
	return
}
